package com.weizhu.swufer.core.registry;

import cn.hutool.core.collection.CollUtil;
import com.google.gson.Gson;
import com.weizhu.swufer.constant.SwuferConstant;
import com.weizhu.swufer.core.entity.RpcBaseInfo;
import com.weizhu.swufer.core.grpc.RemoteCacheManager;
import com.weizhu.swufer.core.exception.RpcException;
import com.weizhu.swufer.core.util.ClassUtil;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class ZkServiceRegister implements ServiceRegister{


    private CuratorFramework cf;

    private String registerCenterAddress;

    private RemoteCacheManager remoteCacheManager;

    private final static String DEFAULT_ROOT = SwuferConstant.ZK_DEFAULT_ROOT;

    private final static Logger logger = LoggerFactory.getLogger(ZkServiceRegister.class);

    private final static ExecutorService registerThreadPool = new ThreadPoolExecutor(5, 5, 0L,
            TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new DefaultThreadFactory("register-service-pool"));

    public void register(List<RpcBaseInfo> rpcBaseInfoList){

        rpcBaseInfoList.forEach(
                rpcBaseInfo -> CompletableFuture.runAsync(()->doRegister(rpcBaseInfo),registerThreadPool)
                        .exceptionally((r)->{
                            logger.error("",r);
                            return null;
                        }).thenRun(()->{
                            logger.info("register {}::{}:{} in zookeeper",rpcBaseInfo.getService(),
                                    rpcBaseInfo.getIp(),rpcBaseInfo.getPort());
                        })
        );
    }

    public List<String> discover(String serviceName) {
        return discover(serviceName,true);
    }

    public List<String> discover(String serviceName,boolean userCache){
        List<String> rpcInfoList = null;
        try {
            if (userCache){
                rpcInfoList = remoteCacheManager.getCacheAddressList(serviceName);
                if (!CollUtil.isEmpty(rpcInfoList)){
                    return rpcInfoList;
                }
            }
            rpcInfoList = cf.getChildren().forPath("/" + serviceName);

        }catch (Exception e){
            logger.error("",e);
        }
        if (CollUtil.isEmpty(rpcInfoList)){
            throw new RpcException(RpcException.REMOTE_SERVICE_NOT_FOUND,"no remote address be found");
        }
        return rpcInfoList;
    }


    private void doRegister(RpcBaseInfo rpcBaseInfo){
        Gson gson = new Gson();
        String info = gson.toJson(rpcBaseInfo);
        String serviceName = rpcBaseInfo.getService();

        String parentPath = "/" + serviceName;
        try {
            if(cf.checkExists().forPath(parentPath) == null){
                cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/" + serviceName);
            }
            String childPath = parentPath + "/" + info;
            if (cf.checkExists().forPath(childPath) != null){
                cf.delete().forPath(childPath);
            }
            cf.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(childPath);

            cf.setData().forPath(childPath,info.getBytes(StandardCharsets.UTF_8));
        } catch (Exception e) {
            logger.error("",e);
        }
        return;
    }
    public ZkServiceRegister(Properties properties){
        String zkAddress = properties.getProperty(SwuferConstant.PROPERTIES_REGISTRY_ADDRESS);

        cf = CuratorFrameworkFactory.builder().connectString(zkAddress)
                .sessionTimeoutMs(20000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .namespace(DEFAULT_ROOT).build();
        cf.start();
        logger.info("zookeeper connect successfully");
        this.remoteCacheManager = RemoteCacheManager.getInstance();
        initCacheService();
    }

    public ZkServiceRegister(String address){
        this.cf = CuratorFrameworkFactory.builder().connectString(address)
                .sessionTimeoutMs(2000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .namespace(DEFAULT_ROOT).build();
        this.cf.start();
        this.remoteCacheManager = RemoteCacheManager.getInstance();
        logger.info("zookeeper connect successfully");
        initCacheService();
    }

    private void initCacheService(){
        List<String> remoteServiceList;
        String root = "/";
        try {
            remoteServiceList = cf.getChildren().forPath(root);
        }catch (Exception e){
            logger.error("can't find service in zk default root:{}",SwuferConstant.ZK_DEFAULT_ROOT,e);
            return;
        }
        for (String remoteService : remoteServiceList) {
            cacheService(remoteService);
        }

    }

    private void cacheService(String serviceName){
        CuratorCache cache = CuratorCache.build(cf, "/" + serviceName);

        PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
            /**
             * <B>方法名称：</B>监听子节点变更<BR>
             * <B>概要说明：</B>新建、修改、删除<BR>
             */
            @Override
            public void childEvent(CuratorFramework cf,
                                   PathChildrenCacheEvent event) {
                if (event == null || event.getData() == null){
                    return;
                }
                String path = event.getData().getPath();
                String serviceInfo = ClassUtil.getServiceAddress(path);
                switch (event.getType()) {
                    case CHILD_ADDED:
                        remoteCacheManager.cacheAddress(serviceName, serviceInfo);
                        logger.info("CHILD_ADDED :"
                                + path);
                        break;
                    case CHILD_REMOVED:
                        remoteCacheManager.removeCacheAddress(serviceName, serviceInfo);
                        logger.info("CHILD_REMOVED :"
                                + path);
                        break;
                    default:
                        break;
                }
            }
        };
        try{

            CuratorCacheListener listener = CuratorCacheListener.builder()
                    .forPathChildrenCache("/" + serviceName,cf,pathChildrenCacheListener).build();
            cache.listenable().addListener(listener);
            cache.start();
        }catch (Exception e){
            logger.error("",e);
        }
    }


}
